package com.xnyzc.lhy.order.mq.consumer;

import com.alibaba.fastjson.JSON;
import com.xnyzc.lhy.common.component.config.security.RocketConfig;
import com.xnyzc.lhy.common.component.redis.RedisCommon;
import com.xnyzc.lhy.common.constant.MqConstant;
import com.xnyzc.lhy.common.constant.MqConsumerGroupConstant;
import com.xnyzc.lhy.common.penum.EDispatchType;
import com.xnyzc.lhy.common.util.CheckUtil;
import com.xnyzc.lhy.order.entity.system.OaSysDispatchLog;
import com.xnyzc.lhy.order.service.system.IOaSysDispatchLogService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;

@Slf4j
@Component
public class MqDispatchRecordConsumer {
    private DefaultMQPushConsumer consumer;

    @Autowired
    RocketConfig config;

    @Autowired
    RedisCommon redisCommon;

    @Value("${rocketMq.isStart}")
    private boolean isStart;

    /**
     * 订单调度记录表
     */
    @Autowired
    IOaSysDispatchLogService dispatchLogService;

    /**
     * 开启监听
     */
    @PostConstruct
    public void init() {
        try {
            if (!isStart) {
                return;
            }
            consumer = new DefaultMQPushConsumer(MqConsumerGroupConstant.DISPATCH_RECORD);
            consumer.setNamesrvAddr(config.getNameServerAddr());
            consumer.subscribe(config.getTopic(), MqConstant.ORDER_DISPATCH_LOG);
            consumer.setConsumeMessageBatchMaxSize(1);
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    MessageExt messageExt = msgs.get(0);
                    String jsonStr = new String(messageExt.getBody());
                    return MqDispatchRecordConsumer.this.dealBody(jsonStr);
                }
            });
            consumer.start();
            log.info("---------订单调度日志消费者启动成功-----o(*^▽^*)┛-------");
        } catch (Exception e) {
            log.info("---------订单调度日志消费者启动失败----(# ﾟДﾟ)--------");
        }
    }

    public ConsumeConcurrentlyStatus dealBody(String jsonStr) {
        try {
            //json内容不为空进入
            if (CheckUtil.strIsNotEmpty(jsonStr)) {
                OaSysDispatchLog logJsonBen = JSON.parseObject(jsonStr, OaSysDispatchLog.class);
                if (CheckUtil.objIsEmpty(logJsonBen)) {
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                log.info("当前要记录的调度日志信息: {},订单号:{}, 日志状态:{}", jsonStr, logJsonBen.getOrderId(), logJsonBen.getType());
                // 1. 创建订单
                OaSysDispatchLog dispatchLog = getOaSysOrderDispatchRecord(logJsonBen.getOrderId());
                dispatchLog.setOrderId(logJsonBen.getOrderId());
                if (logJsonBen.getType().equals(EDispatchType.createOrderBefore.getType())) {
                    dispatchLog.setReceiveOrderTimestamp(System.currentTimeMillis());
                    if (!dispatchLogService.save(dispatchLog)) {
                        log.info("添加调度日志!");
                    }
                }
                // 创建订单后
                else if (logJsonBen.getType().equals(EDispatchType.createOrderAfter.getType())) {
                    dispatchLog.setCreatedOrderTimestamp(System.currentTimeMillis());
                    dispatchLog.setDdOpenId(logJsonBen.getDdOpenId());
                    dispatchLog.setPassengerLng(Double.valueOf(logJsonBen.getPassengerLng()));
                    dispatchLog.setPassengerLat(Double.valueOf(logJsonBen.getPassengerLat()));
                    if (!dispatchLogService.updateById(dispatchLog)) {
                        log.info("创建订单后 更新调度日志!");
                    }
                }
                // 调度开始
                else if (logJsonBen.getType().equals(EDispatchType.dispatchDriverBefore.getType())) {
                    dispatchLog.setDispatchStartTimestamp(System.currentTimeMillis());
                    if (!dispatchLogService.updateById(dispatchLog)) {
                        log.info("调度开始 更新调度日志!");
                    }
                }
                //调度结束
                else if (logJsonBen.getType().equals(EDispatchType.dispatchDriverAfter.getType())) {
                    dispatchLog.setDispatchEndTimestamp(System.currentTimeMillis());
                    Long driverId = logJsonBen.getDriverId();
                    if (CheckUtil.objIsNotEmpty(driverId)) {
                        dispatchLog.setDriverId(Long.valueOf(driverId));
                    }
                    dispatchLog.setDriverLat(logJsonBen.getDriverLat());
                    dispatchLog.setDriverLng(logJsonBen.getDriverLng());
                    dispatchLog.setDispatchDistance(logJsonBen.getDispatchDistance());
                    if (!dispatchLogService.updateById(dispatchLog)) {
                        log.info("调度结束 更新调度日志!");
                    }
                }
                // 抢单开始
                else if (logJsonBen.getType().equals(EDispatchType.grabOrdersBefore.getType())) {
                    dispatchLog.setGrabRequestTimestamp(System.currentTimeMillis());
                    if (!dispatchLogService.updateById(dispatchLog)) {
                        log.info("抢单开始 更新调度日志!");
                    }
                }
                // 抢单结束
                else if (logJsonBen.getType().equals(EDispatchType.grabOrdersAfter.getType())) {
                    dispatchLog.setGrabResponseTimestamp(System.currentTimeMillis());
                    dispatchLog.setSyncResultCode(logJsonBen.getSyncResultCode());
                    if (!dispatchLogService.updateById(dispatchLog)) {
                        log.info("抢单结束 更新调度日志!");
                    }
                }
                // 订单同步
                else if (logJsonBen.getType().equals(EDispatchType.orderSynchronize.getType())) {
                    dispatchLog.setAsyncNoticeGrapResultTimestamp(System.currentTimeMillis());
                    dispatchLog.setAsyncNoticeGrapResult(logJsonBen.getAsyncNoticeGrapResult());
                    if (!dispatchLogService.updateById(dispatchLog)) {
                        log.info("订单同步 更新调度日志!");
                    }
                }

            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            log.error(e.getMessage());
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    public OaSysDispatchLog getOaSysOrderDispatchRecord(String orderNo) {
        OaSysDispatchLog byId = dispatchLogService.getById(orderNo);
        if (CheckUtil.objIsNotEmpty(byId)) {
            return byId;
        }
        return new OaSysDispatchLog();
    }
}
